package com.xxx.rpc.registry.curator;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.NoNodeException;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.WatchedEvent;

import com.xxx.rpc.registry.zookeeper.Constant;
 

public class CuratorZookeeperClient implements StateListener,ZookeeperClient {

	private final CuratorFramework client;
	
	public CuratorZookeeperClient(String zkAddress) {
		//String url = Constant.ZK_REGISTRY_PATH;
		Builder builder = CuratorFrameworkFactory.builder()
				.connectString(zkAddress)
				.retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000))
				.connectionTimeoutMs(Constant.ZK_CONNECTION_TIMEOUT)
                .sessionTimeoutMs(Constant.ZK_SESSION_TIMEOUT);
		String authority = Constant.ZK_AUTHORITY;
		if (authority != null && authority.length() > 0) {
			builder = builder.authorization("digest", authority.getBytes());
		}
		client = builder.build();
		client.getConnectionStateListenable().addListener(
				new ConnectionStateListener() {
					public void stateChanged(CuratorFramework client,
							ConnectionState state) {
						CuratorZookeeperClient.this.stateChanged(state);
					}
				});
		client.start();
	}

	public void stateChanged(final ConnectionState state) {
		System.out.println();
	}

	public void createPersistent(String path) {
		try {
			client.create().forPath(path);
		} catch (NodeExistsException e) {
		} catch (Exception e) {
			throw new IllegalStateException(e.getMessage(), e);
		}
	}

	public void createEphemeral(String path) {
		try {
			client.create().withMode(CreateMode.EPHEMERAL).forPath(path);
		} catch (NodeExistsException e) {
		} catch (Exception e) {
			throw new IllegalStateException(e.getMessage(), e);
		}
	}

	public void delete(String path) {
		try {
			client.delete().forPath(path);
		} catch (NoNodeException e) {
		} catch (Exception e) {
			throw new IllegalStateException(e.getMessage(), e);
		}
	}

	public List<String> getChildren(String path) {
		try {
			return client.getChildren().forPath(path);
		} catch (NoNodeException e) {
			return null;
		} catch (Exception e) {
			throw new IllegalStateException(e.getMessage(), e);
		}
	}

	public boolean isConnected() {
		return client.getZookeeperClient().isConnected();
	}

	public void doClose() {
		client.close();
	}

	private class CuratorWatcherImpl implements CuratorWatcher {

		private volatile ChildListener listener;

		public CuratorWatcherImpl(ChildListener listener) {
			this.listener = listener;
		}

		public void unwatch() {
			this.listener = null;
		}

		public void process(WatchedEvent event) throws Exception {
			if (listener != null) {
				listener.childChanged(event.getPath(), client.getChildren()
						.usingWatcher(this).forPath(event.getPath()));
			}
		}
	}

	public CuratorWatcher createTargetChildListener(String path,
			ChildListener listener) {
		return new CuratorWatcherImpl(listener);
	}

	public List<String> addTargetChildListener(String path,
			CuratorWatcher listener) {
		try {
			return client.getChildren().usingWatcher(listener).forPath(path);
		} catch (NoNodeException e) {
			return null;
		} catch (Exception e) {
			throw new IllegalStateException(e.getMessage(), e);
		}
	}

	public void removeTargetChildListener(String path, CuratorWatcher listener) {
		((CuratorWatcherImpl) listener).unwatch();
	}

	
	
	
	private final Set<StateListener> stateListeners = new CopyOnWriteArraySet<StateListener>();

	private final ConcurrentMap<String, Set<ChildListener>> childListeners = new ConcurrentHashMap<String, Set<ChildListener>>();

	private volatile boolean closed = false;
	
	
	
	public void create(String path, boolean ephemeral) {
		int i = path.lastIndexOf('/');
		if (i > 0) {
			create(path.substring(0, i), false);
		}
		if (ephemeral) {
			createEphemeral(path);
		} else {
			createPersistent(path);
		}
	}

	public void addStateListener(StateListener listener) {
		stateListeners.add(listener);
	}

	public void removeStateListener(StateListener listener) {
		stateListeners.remove(listener);
	}

	public Set<StateListener> getSessionListeners() {
		return stateListeners;
	}

	public void addChildListener(String path, final ChildListener listener) {
		Set<ChildListener> listeners = childListeners.get(path);
		if (listeners == null) {
			childListeners.putIfAbsent(path, new HashSet<ChildListener>());
			listeners = childListeners.get(path);
		}
		listeners.add(listener);
		 
		//return addTargetChildListener(path, listener);
	}

	public void removeChildListener(String path, ChildListener listener) {
		Set<ChildListener> listeners = childListeners.get(path);
		if (listeners == null) {
			return;
		}
		if(listeners.contains(listener)){
			listeners.remove(listener);
		}
	}

	 

	public void close() {
		if (closed) {
			return;
		}
		closed = true;
		try {
			doClose();
		} catch (Throwable t) {
			//LOGGER.warn(t.getMessage(), t);
		}
	}

}
